package org.looop.chuck.schedule.impl;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
import org.apache.commons.httpclient.params.HttpConnectionManagerParams;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.looop.chuck.http.RequestHandler;
import org.looop.chuck.parser.HTMLParserFactory;
import org.looop.chuck.persistence.PersistenceManager;
import org.looop.chuck.schedule.URLTaskScheduler;
import org.looop.chuck.task.TaskFailedReason;
import org.looop.chuck.task.URLTaskManager;

public class DefaultScheduler implements URLTaskManager, URLTaskScheduler {
	private final Log log = LogFactory.getLog(this.getClass());
	private ConcurrentHashMap<String, Object> doingTaskSet = new ConcurrentHashMap<String, Object>();
	private Map<String, TaskFailedReason> failedTasks = new ConcurrentHashMap<String, TaskFailedReason>();
	private ExecutorService exec;
	private HttpClient httpClient;
	private ConcurrentHashMap<String, Object> doneTaskSet = new ConcurrentHashMap<String, Object>();
	private PersistenceManager persistenceManager;
	private Semaphore semaphore;
	private volatile long lastModified = System.currentTimeMillis();
	private final long max_interval_time = 3 * 1000;
	private HTMLParserFactory htmlParserFactory;
	
	public DefaultScheduler(PersistenceManager persistenceManager, ExecutorService exec, HTMLParserFactory htmlParserFactory) throws UnsupportedEncodingException {
		this.exec = exec;
		HttpConnectionManagerParams params = new HttpConnectionManagerParams();
		params.setDefaultMaxConnectionsPerHost(20);
		params.setMaxTotalConnections(20);
		MultiThreadedHttpConnectionManager connectionManager = new MultiThreadedHttpConnectionManager();
		connectionManager.setParams(params);
		httpClient = new HttpClient(connectionManager);
		this.persistenceManager = persistenceManager;
		this.htmlParserFactory = htmlParserFactory;
		this.semaphore = new Semaphore(1);
		try {
			this.semaphore.acquire();
		} catch (InterruptedException e1) {
			e1.printStackTrace();
			return;
		}
		Thread schedulerConsoleMonitor = new Thread(){
			@Override
			public void run() {
				while(true){
					try {
						sleep(3000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					long now = System.currentTimeMillis();
					long interval = now - lastModified;
					if(interval > max_interval_time && doingTaskSet.size() == 0){
						DefaultScheduler.this.semaphore.release();
						break;
					}
					System.out.println("***********************************************");
					System.out.println("doing task size: " + doingTaskSet.size());
					System.out.println("done task size: " + doneTaskSet.size());
					System.out.println("fail task size: " + failedTasks.size());
				}
			}
		};
		schedulerConsoleMonitor.setDaemon(true);
		schedulerConsoleMonitor.start();
	}
	
	public void submit(String url) {
		lastModified = System.currentTimeMillis();
		if(doingTaskSet.containsKey(url) || doneTaskSet.containsKey(url)){
			log.info(url + " task exists");
			return;
		}
		log.info(url + " has been submit");
//		final RequestCallable callable = new RequestCallable(httpClient, url);
//		FutureTask<HttpMethod> futureTask = new HTTPRequestTask(callable, this, persistenceManager);
		RequestHandler task = new RequestHandler(httpClient, url, this, persistenceManager, htmlParserFactory);
		doingTaskSet.put(url, "");
		exec.submit(task);
//		new Thread(task).start();
	}
	
	public void awaitFinish(){
		try {
			this.semaphore.acquire();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}



	@Override
	public int addTaskList(List<String> urlList) throws Exception {
		if(urlList != null && urlList.size() > 0){
			int i = 0;
			for (String url : urlList) {
				submit(url);
				i++;
			}
			return i;
		}
		return 0;
	}

	@Override
	public void successTask(String url) {
//		if(doingTaskSet.containsKey(url)){
			log.info(url + " has been success processed");
			doingTaskSet.remove(url);
			log.info(url + " has been success removed");
			doneTaskSet.put(url, "");
			log.info(url + " has been success put done task");
//		}else{
//			System.out.println(url + " not exsists");
//		}
	}

	@Override
	public void failedTask(String url, TaskFailedReason reason) {
//		if(doingTaskSet.containsKey(url)){
			log.info(url + " has been failed, " + reason.toString());
			failedTasks.put(url, reason);
			log.info(url + " has been put into failed task");
			doingTaskSet.remove(url);
			log.info(url + " has been success removed");
			doneTaskSet.put(url, "");
			log.info(url + " has been success put done task");
//		}
	}
	
}
